package mysql;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

public class MysqlTable
{
    public static void main(String[] args) throws Exception
    {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv  = TableEnvironment.getTableEnvironment(env);
        DataSet<Row> dbData =
        env.createInput(
                JDBCInputFormat.buildJDBCInputFormat()
                        .setDrivername(ConfigKeys.DRIVER_CLASS)
                        .setDBUrl(ConfigKeys.SOURCE_DRIVER_URL)
                        .setUsername(ConfigKeys.SOURCE_USER)
                        .setPassword(ConfigKeys.SOURCE_PASSWORD)
                        .setQuery("SELECT goods_id,goods_name FROM goods_info")
                        .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
                        .finish()
        );

        DataSet<Row> dbData1 =
                env.createInput(
                        JDBCInputFormat.buildJDBCInputFormat()
                                .setDrivername(ConfigKeys.DRIVER_CLASS)
                                .setDBUrl(ConfigKeys.SOURCE_DRIVER_URL)
                                .setUsername(ConfigKeys.SOURCE_USER)
                                .setPassword(ConfigKeys.SOURCE_PASSWORD)
                                .setQuery("SELECT order_id,CAST(goods_id AS SIGNED) goods_id FROM goods_order_info_detail")
                                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO))
                                .finish()
                );

        tEnv.registerDataSet("GoodsInfo", dbData, "goods_id, goods_name");
        tEnv.registerDataSet("GoodsOrder", dbData1, "order_id,goods_id");

        Table table = tEnv.sqlQuery(
                "SELECT GoodsOrder.order_id, GoodsInfo.goods_id, GoodsInfo.goods_name as frequency FROM GoodsInfo INNER JOIN GoodsOrder ON GoodsInfo.goods_id = GoodsOrder.goods_id ");


        DataSet<Row> result = tEnv.toDataSet(table, new RowTypeInfo(BasicTypeInfo.LONG_TYPE_INFO,BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO));

        result.output(
                JDBCOutputFormat.buildJDBCOutputFormat()
                        .setDrivername(ConfigKeys.DRIVER_CLASS)
                        .setDBUrl(ConfigKeys.SOURCE_DRIVER_URL)
                        .setUsername(ConfigKeys.SOURCE_USER)
                        .setPassword(ConfigKeys.SOURCE_PASSWORD)
                        .setQuery("INSERT INTO goods_info_sink1(order_id,goods_id,goods_name) values(?,?,?)")
                        .finish()
        );

        result.output(
                JDBCOutputFormat.buildJDBCOutputFormat()
                        .setDrivername(ConfigKeys.DRIVER_CLASS)
                        .setDBUrl(ConfigKeys.SOURCE_DRIVER_URL)
                        .setUsername(ConfigKeys.SOURCE_USER)
                        .setPassword(ConfigKeys.SOURCE_PASSWORD)
                        .setQuery("INSERT INTO goods_info_sink2(order_id,goods_id,goods_name) values(?,?,?)")
                        .finish()
        );

        env.execute("flink mysql Dataset demo");
    }

    public static class GI {
        public long goods_id;
        public String  goods_name;

        // public constructor to make it a Flink POJO
        public GI() {}

    }
}
